use std::sync::Arc;
use std::thread;

use crate::abstractions::data::Data;
use crate::abstractions::task::TaskAbstraction;
use crate::datafile::taskfile::TaskFileLogger;
use crate::error::ServerError;
use crate::log::task_logger::TASK_SUMMARY_LOGGER;
use crate::protocol::type_schema::TypeSchema;
use crate::services::manager::SERVICE_MANAGER;
use crate::services::service::Service;
use crate::protocol::task::{SymbolicTask, TaskProgressD, TaskStatus};
use crate::tasks::manager::TASK_MANAGER;
use crate::tasks::task::{task_get_result, Task};
use crate::protocol::data_schema::DataTrait;
use std::sync::mpsc::TryRecvError;

pub fn workflow_start_task(workflow_port: u16, task_name: Arc<String>, input_data: Data, service: Arc<Service>) -> Result<Data, ServerError> {
    if let Some(task) = service.get_task(&task_name) {
        let handle = thread::spawn(move || {
            let rt = tokio::runtime::Builder::new_current_thread()
                    .enable_all()
                    .build()
                    .unwrap();
            rt.block_on(async {
                portable_task_func(&service, task, input_data, workflow_port).await
            })
        });
        handle.join().unwrap()
    } else {
        Err(ServerError::ServiceError(format!("任务不存在: {}", task_name)))
    }
}

pub async fn portable_task_func(service: &Service, task_sig: SymbolicTask, input: Data, workflow_port: u16) -> Result<Data, ServerError> {
    let mut task = Task::new_from_workflow(service.address.clone(), service.name.clone(), task_sig.name, task_sig.input_type, task_sig.output_type, service.get_session_key(), workflow_port);
    let task_id = task.id.clone();
    let handle = tokio::spawn(async move {
        let instant = task.connect(input).await?;
        if instant {
            match task.status {
                TaskStatus::Completed => return Ok(task.result.unwrap()),
                TaskStatus::Failed => return Err(ServerError::TaskError(format!("远程任务失败: {}", task.error_message.unwrap_or("(无错误信息)".into())))),
                _ => unreachable!()
            }
        }
        let mut progress_receiver = task.progress_receiver.take();
        let mut log_receiver = task.log_receiver.take();
        let task_id = task.id.clone();
        let terminate_flag = task.terminate_flag.clone();
        let info_str = format!("{} - {} - {}", task.service_name, task.name, task.id);
        let mut manager = TASK_MANAGER.get().unwrap().lock().await;
        manager.add_task(task)?;
        drop(manager);
        let mut last_update = std::time::Instant::now();
        let exit_status ;
        'main_loop: loop {
            if *terminate_flag.lock().unwrap() {
                exit_status = TaskStatus::Stopped;
                break 'main_loop;
            }
            if let Some((ref mut receiver, _)) = progress_receiver {
                match receiver.try_recv() {
                    Ok(progress_str) => {
                        if progress_str == "Closed" {
                            exit_status = TaskStatus::Closed;
                            break 'main_loop;
                        }
                        match serde_json::from_str::<TaskProgressD>(&progress_str) {
                            Ok(progress) => {
                                // task in workflow 每2秒更新一次 taskmanager
                                if last_update.elapsed() > std::time::Duration::from_secs(2) {
                                    let mut manager = TASK_MANAGER.get().unwrap().lock().await;
                                    let task = match manager.get_task_mut(&task_id) {
                                        Some(v) => v,
                                        None => return Err(ServerError::TaskErrorStr("任务不存在"))
                                    };
                                    task.update_progress(progress.status, progress.percentage, progress.errmsg);
                                    drop(manager);
                                    last_update = std::time::Instant::now();
                                }
                                match progress.status {
                                    TaskStatus::Completed | TaskStatus::Failed | TaskStatus::Stopped => {
                                        exit_status = progress.status;
                                        break 'main_loop;
                                    },
                                    TaskStatus::Running | TaskStatus::Created => {}
                                    other => {
                                        return Err(ServerError::TaskError(format!("远程不应该返回此状态：{:?}", other)));
                                    }
                                }
                            },
                            Err(e) => {
                                return Err(ServerError::TaskError(format!("解析进度信息失败: {}, 原始数据: {}", e, progress_str)));
                            }
                        }
                    },
                    Err(TryRecvError::Disconnected) => return Err(ServerError::TaskErrorStr("通道断开")),
                    _ => {}
                }
            }
            if let Some((ref mut receiver, _)) = log_receiver {
                if let Ok(log) = receiver.try_recv() {
                    // task.log(log);
                }
            }
            std::thread::sleep(std::time::Duration::from_millis(200));
        }

        if let Some((_, switch)) = progress_receiver {
            switch.store(false, std::sync::atomic::Ordering::Relaxed);
        }
        if let Some((_, switch)) = log_receiver {
            switch.store(false, std::sync::atomic::Ordering::Relaxed);
        }
        // println!("check1 {}", task_id);
        // 标记退出状态
        if let Ok(mut manager) = TASK_MANAGER.get().unwrap().try_lock() {
            let task = manager.get_task_mut(&task_id).unwrap();
            match exit_status {
                TaskStatus::Completed => {
                    task.set_status(exit_status);
                },
                TaskStatus::Failed | TaskStatus::Stopped | TaskStatus::Closed => {
                    task.end_set_status(exit_status);
                    match exit_status {
                        TaskStatus::Failed => {
                            return Err(ServerError::TaskErrorStr("任务失败"));
                        }
                        TaskStatus::Stopped => {
                            return Err(ServerError::TaskErrorStr("任务手动停止"));
                        }
                        TaskStatus::Closed => {
                            return Err(ServerError::TaskErrorStr("远程连接关闭"));
                        }
                        _ => unreachable!()
                    }
                }
                _ => unreachable!()
            }
        }
        else {
            println!("任务管理器被锁，更新任务状态失败: {}", task_id);
        }
        // println!("check2 {}", task_id);
        // 获取结果，不能被锁否则可能死锁
        let task_id_clone = task_id.clone();
        let get_res = task_get_result(task_id_clone).await;
        let res = match get_res {
            Ok(result) => {
                let mut manager = TASK_MANAGER.get().unwrap().lock().await;
                let task = manager.get_task_mut(&task_id).unwrap();
                task.end_set_status(TaskStatus::ResultReceived);
                drop(manager);
                Ok(result)
            }
            Err(e) => {
                let mut manager = TASK_MANAGER.get().unwrap().lock().await;
                let task = manager.get_task_mut(&task_id).unwrap();
                task.end_set_status(TaskStatus::FailedToGetResult);
                drop(manager);
                Err(e)
            }
        };
        // println!("check3 {}", task_id);
        // 日志和清除任务记录
        let handle = std::thread::spawn(move || {
            let rt = tokio::runtime::Builder::new_current_thread()
                    .enable_all()
                    .build()
                    .unwrap();
            rt.block_on(async {
                let _ = mark_for_log_and_clear_record(task_id).await;
            })
        });
        res
    });
    handle.await.unwrap()
}

async fn mark_for_log_and_clear_record(task_id: String) -> Result<(), ServerError> {
    let max_retries = 5;
    let interval_secs = 5;
    let mut retry = 0;
    let mut success = true;
    thread::sleep(std::time::Duration::from_secs(5));
    loop {
        if let Ok(manager) = TASK_MANAGER.get().unwrap().try_lock() {
            let task = manager.get_task(&task_id).unwrap();
            let logger = TASK_SUMMARY_LOGGER.lock().unwrap();
            if let Err(e) = logger.log_task(task) {
                println!("任务日志记录失败: {}", e);
            }
            let file_saver = TaskFileLogger::new();
            file_saver.save_task_result(task)?;
            break;
        }
        else {
            println!("任务管理器被锁……");
            retry += 1;
            if retry >= max_retries {
                println!("重复{}次拿不到锁，任务日志记录失败: {}", retry, task_id);
                success = false;
                break;
            }
        }
        thread::sleep(std::time::Duration::from_secs(interval_secs));
    }

    thread::sleep(std::time::Duration::from_secs(60));
    let mut retry = 0;
    loop {
        if let Ok(mut manager) = TASK_MANAGER.get().unwrap().try_lock() {
            manager.remove_task(&task_id);
            break;
        }
        else {
            println!("任务管理器被锁……");
            retry += 1;
            if retry >= max_retries {
                println!("重复{}次拿不到锁，清除任务失败: {}", retry, task_id);
                success = false;
                break;
            }
        }
        thread::sleep(std::time::Duration::from_secs(interval_secs));
    }

    if success {
        Ok(())
    }
    else {
        Err(ServerError::TaskErrorStr("任务收尾失败"))
    }
}